Companies today are tasked with implementing solutions for many types of integration challenges within their respective enterprises. Many of these challenges involve issues of application integration (e.g., integration among and/or between software applications and/or other systems) and fall into common patterns.
For example, a first area relates to propagation of similar business objects from one system to multiple other systems, such as, for example, in an order status change or a product price change. A second area relates to synchronization of similar business objects between two or more systems to obtain a single view, such as, for example, a real-time synchronization of customer, product registration, product order, and product SKU information among several applications. This is one of the most common issues requiring an integration solution. In a one-way synchronization, there generally is one system (e.g., resource) that acts as a data source and one or more resources that are targets of the synchronization. In a two-way synchronization, every resource often is both a potential source and target of a synchronization. There generally is not a single resource that acts as the primary data resource. Thus, a change to any resource should be reflected in all other resources. A third area involves information joined from multiple sources into a common destination system, such as, for example, communicating pharmacy customer records and prescription transactions and website data into a central application and database.
Various tools have been provided that enable a user to design and deploy solutions that address these challenges using, for example, the publish-and-subscribe model or one of its variants. The publish-and-subscribe model is a specific type of message-based solution in which messages are exchanged anonymously through a message broker. Applications that produce information that needs to be shared make this information available in specific types of recognizable documents that they publish to the message broker. Applications that require information subscribe to the document types they need.
At runtime, the message broker receives documents from publishers and then distributes the documents to subscribers. The subscribing application processes or performs work using the document and may or may not send a response to the publishing application.
In a typical system, an integration server or applications running on an integration server publish documents to a broker. The broker then routes the documents to subscribers located on other integration servers. The integration server and the broker share a fast, efficient process for exchanging documents across the entire system.
Brokers can be linked to form units known as territories, and brokers that form a territory may sometimes function as a single logical broker. Clients that connect to one broker in a territory can automatically exchange documents with clients connected to any of the brokers in the territory. Territories can be organized in any desired way, e.g., to take into account factors such as, for example, geographical distribution of clients and/or back-end services, the types of services offered, actual and/or expected traffic, actual and/or expected processing loads, etc. Territories thus may help support scalability by allowing one to segment traffic among multiple brokers, while still treating the brokers as a single operational unit.
Alternatively, or in addition, brokers can be grouped into clusters. As used herein and is understood by those skilled in the art, territories and clusters are different concepts. Clusters may be thought of as including a set of loosely or tightly connected computers (depending on the implementation) that work together so that they can in many respects be viewed as a single system. Territories as described herein typically will have no master/slave relationship, whereas the same is not necessarily true when clusters are involved. In this vein, one further difference as between territories and clusters may relate to the way in which clusters oftentimes are set up and run (e.g., apart from the master/slave concept) and, more particularly, to the notion of typically using clusters to distribute workload and solve performance and/or reliability issues with a system, as these concepts generally are not associated with what a territory does. For instance, a territory may use a set of brokers to distribute the workload, and also frequently may involve a message, document, or the like, being published to all subscribers across all brokers at the same time, e.g., such that a publisher client only has to publish to a nearest (or any) broker, with the system ensuring that subscriber clients in the group (allowed to receive the publication) can access it from its nearest (or any) broker in the territory (or even through a territory gateway to another territory). By contrast, a cluster might in some cases help to distribute the workload amongst the brokers in a cluster, but involve a message, document, or the like, being published to only one broker in a cluster.
It will be appreciated that it would be desirable in many instances (including, for example, those where mission-critical applications are involved) to provide a messaging infrastructure technology that supports the capability to send data from a producer to a consumer in the fastest possible manner while still ensuring guaranteed delivery. The transmission of documents using technologies like the Java Message Service (JMS), for example, has become common in data exchange/brokerage scenarios. Significant performance boosts can be observed in the case of volatile (e.g., non-persisted) data. But volatile data, while at least in theory promising a boost to the overall performance of the data transmission, tends to carry with it a cost of non-reliability (e.g., as the failover case is not handled). As a result, in such instances, the tradeoff of reliability versus performance is weighted in the favor of performance in many technological implementations.
To help guarantee delivery (in the event of a system failure, for example), messaging systems oftentimes recommend administrators to mark the messages as “persistent.” In many common messaging systems, a persistent message will be guaranteed to be delivered once-and-only-once. The general goals in such cases are that persistent messages will not be lost as a result of a messaging system failure or the like, and that they will not be delivered two or more times. When operating in a persistent mode, a persistent message generally will not be considered sent until it has been safely written to a file or database.
But resorting to a “persistence mode” can have adverse effects on message transmission performance. As will be appreciated, the costs of ensuring reliability generally will become greater when transmission reliability is to be guaranteed and/or when the data being transmitted is large. For instance, hard disk I/O operations for persistent messages can create bottlenecks, e.g., as the system may need to write each message that is received to the file system to persist the message. The message throughput and how fast one can get more data transferred from the producer to the consumer thus can depend greatly on file system I/O. Moreover, in the case of JMS, for example, persisting a Java message instance to a persistent storage may require the instance to be “serialized,” e.g., to a binary form. Java serialization comes to mind immediately, but as versatile and automated it may be (by virtue of being reflection based), it maybe far from optimal from a pure performance perspective.
When it comes to transferring large data sets, for example, marking the data as persistent also can affect the data throughput significantly, e.g., as the potentially resource-intensive activities described above may take yet longer when yet more data is involved. Large amounts of data may present challenges, even for data transmissions in non-persistent modes of operation. That is, data transmission via a broker in a non-persistent mode may require the data in transit to be in memory. Although the data oftentimes can be live-streamed from the sending entity to the broker queue, which then can be picked up by the consuming entity from the broker queue, additional challenges may be presented as the size of the data increases. For instance, this approach may have serious repercussions, as a broker node failing may lead to the potential loss of data in transit. Moreover, a consuming entity picking up data at a slower rate (e.g., as more and more data needs to be handled) could inundate the broker queue with data and potentially exhaust the runtime memory, cause the broker to go down, etc.
Some messaging middleware providers have in the past provided advanced features like active/active clustering (e.g., where traffic intended for a failed node is either passed on to an existing node or load-balanced across remaining nodes) to help scale-up data throughput. Some of these advanced features, including JMS clustering, are in many ways similar to modern distributed systems that adopt a “shared nothing architecture” which, as will be appreciated, may in essence be an architecture in which each node is independent and self-sufficient, and where there is no single point of contention across the system. More specifically, in at least some of such cases, none of the nodes share memory or disk storage.
The webMethods JMS Policy-Based Clustering is one example where the message distribution load is shared across multiple instances of the broker (message queue), each of which has its own memory and disk storage. See, for example, U.S. Pat. No. 8,453,163, the entire content of which is hereby incorporated herein by reference.
Unfortunately, the use a cluster of data brokers has not completely resolved the tensions between performance and reliability. Indeed, although the JMS Policy-Based Clustering techniques provided by the assignee represent a good attempt to achieve active/active clustering and load balancing (e.g., in accordance with a round robin algorithm, or the like) for faster throughput and better scalability, further improvements may be possible. For example, data may still need to be marked as persistent to ensure reliability. As another example, message-level failover capability, e.g., in the case where any broker node processing a part of the message goes down because of a system or application level failure or the like, is not necessarily provided. This may present a drawback in use cases where, for example, at least once message delivery is a strict business requirement.
Thus, it will be appreciated that there is a need in the art for improved data transmission techniques, e.g., where reliability is desired in connection with large data sets.
One aspect of certain example embodiments relates to techniques for including intelligence into a messaging system to dynamically and automatically tune a data transmission process at runtime, e.g., in helping to improve throughput without compromising on reliability.
In certain example embodiments, a messaging infrastructure technology is provided that enables large data sets to be sent with a very high throughput while still ensuring guaranteed delivery. Message throughput is scaled by not marking the messages as persistent, thus helping to avoid disk I/O operations. However, message reliability is improved by implementing logic that helps ensures that even if a broker goes down, the data present in that broker queue is still delivered to the consumer.
In certain example embodiments, a method of transmitting data from a sending entity to a receiving entity over a computer-mediated network via a brokering entity is provided. The computer-mediated network includes a plurality of brokers. Fragments of a file to be transmitted are received from the sending entity and at the brokering entity. The method includes, for each said fragment: dynamically determining first and second brokers to which the respective fragment is to be sent, the first and second brokers being different from one another and each having primary and backup queues, and the dynamic determinations being variable for different fragments; storing the respective fragment to transitory memory stores of the first and second brokers; adding to the primary queue of the first broker an indicator of the respective fragment; adding to a backup queue of the second broker the indicator of the respective fragment; and transmitting the received fragment from the first broker whenever possible, and otherwise transmitting the received fragment from the second broker, for reassembly of the file at the receiving entity. Determinations of the first and second brokers to which the fragments are to be sent are based on indications of health for each of the brokers in the network, with the indications of health being fed back to the sending entity through the brokering entity. The sending, receiving, and brokering entities each are computer devices that include respective processing resources including processors and memories.
According to certain example embodiments, the brokers may be organized in one or more clusters, and brokers optionally may be added to a given cluster based on indications of health for the brokers in the given cluster.
According to certain example embodiments, the indications of health for each of the brokers in the network are based at least in part on how full the primary queues on the respective brokers are. For instance, each indication of health may be classified as being one of three different levels of healthiness such that, for example, a first level of healthiness corresponds to normal broker operation, a second level of health indicates that new fragments at least temporarily should be sent to the respective broker with a reduced regularity, and a third level of health indicates that new fragments at least temporarily should not be sent to the respective broker.
According to certain example embodiments, indicators may be removed from backup queues in accordance with a predefined eviction policy. In this regard, the indicator of the fragment(s) currently being transmitted may in certain example embodiments be persisted in a store of the brokering entity. When a received fragment cannot be transmitted from the first broker, certain example embodiments may involve: retrieving, from the store, the indicator of the received fragment that cannot be transmitted from the first broker; removing any entries in the backup queue of the second broker that come before the indicator of the received fragment that cannot be transmitted from the first broker; and transmitting fragments from the backup queue of the second broker, e.g., until the first broker is able to transmit fragments once again.
According to certain example embodiments, determinations of the first and second brokers to which the fragments are to be sent may be made at the brokering entity, e.g., based on gates through which the individual fragments are received. Similarly, according to certain example embodiments, determinations of the first and second brokers to which the fragments are to be sent may be made at the sending entity, e.g., based on broker health related information transmitted thereto.
According to certain example embodiments, fragment transmissions may be mediated via the publish-subscribe model, or one of its variants.
According to certain example embodiments, plural sending devices and/or receiving devices may be provided.
According to certain example embodiments, fragment size and/or fragment transmission frequency to a given channel may be overridable based on health indicators.
According to certain example embodiments, the file and its fragments may be treated by the brokering entity and the brokers as volatile messages and/or neither the file nor the fragments thereof is marked as persistent.
According to certain example embodiments, at least one receiving device may be configured to (a) assemble received fragments in reconstructing the file and/or (b) perform duplicate fragment detection.
In certain example embodiments, a transmission system for use in a computer-mediated network. At least one sending device comprises a first processor and a first memory, and at least one receiving device comprises a second processor and a second memory. At least one brokering entity to which a plurality of broker devices are connected is provided. The broker devices include respective processors and memories and implement respective primary and backup queues. The at least one brokering entity is connected to the at least one sending device via a plurality of channels. The at least one sending device is configured to use its first processor to at least: divide a file to be transmitted into a plurality of fragments, and for each said fragment, dynamically determine which channel to use in transmitting the respective fragment to the at least one brokering entity, with the channel determination being based on health indicators of the broker devices relied to the at least one sending device via the at least one brokering entity. The at least one brokering entity is configured to at least: receive fragments from the at least one sending device via gates respectively connected to the channels, send received fragments to first and second broker devices in dependence on the gate through which the received fragments were received and cause indicators of the received fragments to be enqueued in the primary queues of the first broker devices and backup queues of the second broker devices, and cause received fragments to be transmitted to the at least one receiving device from the first broker devices whenever possible, and otherwise cause received fragments to be transmitted to the at least one receiving device from the corresponding second broker devices.
In certain example embodiments, a brokering entity is configured to relay files among computerized sending and receiving devices through a plurality of brokers deployed in a computer network. The brokering entity comprises at least processor and a memory, a persistent data store, and a plurality of gates to which channels accessibly by the sending devices are connected. The at least one processor and the memory cooperate in managing a data flow controller provided to the brokering entity, the data flow controller being configured to: receive, from the sending entity and through the gates, fragments of a file to be transmitted; and for each said fragment: select first and second brokers from the plurality of brokers based on the gate through which the respective fragment was received, enqueue an identifier of the respective fragment to a primary queue of the selected first broker and a backup queue of the selected second broker, cause the respective fragment to be stored in transitory memory stores of the first and second brokers, cause the received fragment to be transmitted from the first broker to one or more intended recipient receiving devices whenever possible, and otherwise transmitting the received fragment from the second broker thereto, store indicators of the fragments being transmitted to the persistent data store, and transmit broker health related information to the sending devices in order to aid in selecting the channel over which a given fragment is to be transmitted.
In certain example embodiments, there are provided one or more non-transitory computer readable storage mediums tangibly storing instructions that, when executed by at least one processor of a system, perform the above-described and/or other methods.
These aspects and example embodiments may be used separately and/or applied in various combinations to achieve yet further embodiments of this invention.